package rpc.center;

import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lombok.extern.java.Log;
import lombok.extern.slf4j.Slf4j;
import rpc.common.NodeInformation;
import rpc.common.TaskRpc;
import rpc.units.SocketUnits;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;

@Log
public class CenterServiceImpl implements ICenterService {

    private ScheduledExecutorService jumpThreadPool = Executors.newScheduledThreadPool(1);
    private ExecutorService threadPool = Executors.newFixedThreadPool(3);
    private ServerSocket serverSocket;
    private Map<String, NodeInformation> serverNodeInformationMap =new HashMap<>();
    private Map<String, NodeInformation> errorServerNodeInformationMap =new HashMap<>();

    private volatile boolean isRunning;

    @Override
    public void register(NodeInformation nodeInformation) {
        log.log(Level.INFO,"服务注册:"+nodeInformation.getNodeName());
        serverNodeInformationMap.put(nodeInformation.getNodeName(),nodeInformation);
        //移除坏节点
        errorServerNodeInformationMap.remove(nodeInformation.getNodeName());
    }

    @Override
    public void start() {
        isRunning=true;
        initCenter();
    }

    @SneakyThrows
    @Override
    public void stop() {
        //经过实验发现,如果close()时刻还有读写阻塞,那么会进行完毕。
        isRunning=false;
        this.serverSocket.close();
        this.serverNodeInformationMap.clear();//清空已经注册的连接
        this.errorServerNodeInformationMap.clear();
    }

    @Override
    public boolean isAlive() {
        return isRunning;
    }

    @SneakyThrows
    private void initCenter(){

        log.log(Level.INFO,"注册中心开启");

        if(this.serverSocket==null||this.serverSocket.isClosed()) {
            this.serverSocket = new ServerSocket();
            SocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 8888);
            serverSocket.bind(socketAddress);
        }

        jumpThreadPool.execute(this::testServer);

        while (isRunning){

            try {
                Socket socket = serverSocket.accept();
                threadPool.execute(()-> this.executeSocket(socket));
            }catch (Exception e){
                log.log(Level.WARNING,e.getMessage());
                e.printStackTrace();
            }

        }

        log.log(Level.WARNING,"注册中心关闭");

    }

    @SneakyThrows
    private void executeSocket(Socket socket) {

        String receiveMsg = SocketUnits.receiveMsg(socket);//外部手动close
        if(receiveMsg.contains("nodeName")){
            //服务注册信息
            NodeInformation nodeInformation= JSON.parseObject(receiveMsg,NodeInformation.class);
            if(nodeInformation.getStatus()==0){
                //注销服务
                this.serverNodeInformationMap.remove(nodeInformation.getNodeName());
            }else {
                //注册服务
                this.register(nodeInformation);
            }

        }else{
            //客户端获取服务端信息
            TaskRpc taskRpc= JSON.parseObject(receiveMsg,TaskRpc.class);
            String serverName = taskRpc.getServerName();
            NodeInformation nodeInformation = this.serverNodeInformationMap.get(serverName);

            if(nodeInformation==null){
                nodeInformation=this.errorServerNodeInformationMap.get(serverName);
                if(nodeInformation!=null){
                    //去重新获取连接试一试
                    try {
                        String result = callServer(nodeInformation, "jump");
                        //复燃开始
                        nodeInformation = JSON.parseObject(result, NodeInformation.class);
                        this.errorServerNodeInformationMap.remove(nodeInformation.getNodeName());
                        this.serverNodeInformationMap.put(nodeInformation.getNodeName(),nodeInformation);
                    }catch (Exception e){
                        //从坏节点中重新复燃失败
                        log.log(Level.INFO,"节点重新连接:"+nodeInformation.getNodeName());
                    }
                }
            }

            if(nodeInformation==null){
                SocketUnits.sendMsg(socket,"null");
            }else{
                //返回服务端的IP+端口,让客户端自己调用
                SocketUnits.sendMsg(socket,JSON.toJSONString(nodeInformation));
            }
            //客户端获取完就关闭
            this.close(socket);
        }

    }

    private void close(Socket socket) throws IOException {
        socket.close();
    }

    private void testServer() {
        Map<String,NodeInformation> errorServerMap=new HashMap<>();
        for (Map.Entry<String, NodeInformation> entry : serverNodeInformationMap.entrySet()) {
            NodeInformation nodeInformation = entry.getValue();
            //心跳
            try {
                callServer(nodeInformation, "jump");
            }catch (Exception e){
                log.log(Level.WARNING,"一次检测:失去连接");
                //服务端异常,多次检测开始
                errorServerMap.put(entry.getKey(),entry.getValue());
            }
        }
        if(errorServerMap.size()>0){
            //前往多次检测
            this.testServerAgain(errorServerMap);
        }
        //10S检测
        this.jumpThreadPool.schedule(()->testServer(),10, TimeUnit.SECONDS);
    }

    private void testServerAgain(Map<String,NodeInformation> map) {
        for (Map.Entry<String, NodeInformation> entry : map.entrySet()) {
            NodeInformation nodeInformation = entry.getValue();
            //心跳
            try {
                callServer(nodeInformation, "jump");
            }catch (Exception e){
                log.log(Level.WARNING,"二次检测:失去连接");
                //去除当前服务端
                serverNodeInformationMap.remove(entry.getKey());
                //放入坏节点
                errorServerNodeInformationMap.put(entry.getKey(),entry.getValue());
            }
        }
    }

    @SneakyThrows
    private String callServer(NodeInformation nodeInformation,String msg) {
        Socket socket=new Socket();
        SocketAddress socketAddress=new InetSocketAddress(nodeInformation.getIp(),nodeInformation.getPort());
        socket.connect(socketAddress);
        SocketUnits.sendMsg(socket,msg);
        String result = SocketUnits.receiveMsg(socket);
        this.close(socket);
        return result;
    }

    public static void main(String[] args) {
        CenterServiceImpl centerService=new CenterServiceImpl();
//        new Thread(()->{
//            try {
//                Thread.sleep(3000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            centerService.stop();
//            try {
//                Thread.sleep(3000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            centerService.start();
//        }).start();
        centerService.start();
    }

}
